package com.ch1.kafka001.commit;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;

//kafka-topics.bat --zookeeper localhost:2181 --create --topic topic1 --replication-factor 1 --partitions 4

//消费者：异步手动提交当前偏移量，使用生产者ProducerCommit
public class CommitAsync {
    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers","localhost:9092");
        props.put("key.deserializer", StringDeserializer.class);
        props.put("value.deserializer",StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"group02"); //消费者群组

        props.put("enable.auto.commit",false); //取消自动提交

        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);
        try{

            consumer.subscribe(Collections.singletonList("topic1"));
            while(true){
                ConsumerRecords<String,String> records = consumer.poll(500);
                for(ConsumerRecord<String,String> record: records){
                    System.out.println("主题："+record.topic()+",分区："+record.partition()+",偏移量："+
                            record.offset()+",key:"+record.key()+",value:"+record.value());
                }

                consumer.commitAsync(); //异步提交偏移量
                //允许执行回调
                consumer.commitAsync(new OffsetCommitCallback() {
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                        if(e !=null){
                            System.out.println("cimmit failed for offsets");
                            System.out.println(map);
                            e.printStackTrace();
                        }
                    }
                });

            }
        }finally {
            consumer.close();
        }

    }
}
